package mr

import (
	"errors"
	"fmt"
	"io/ioutil"
	"log"
	"net"
	"net/http"
	"net/rpc"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"
)

var mutex sync.Mutex

type Phase int

const (
	mapPhase = iota
	reducePhase
	workDonePhase
)

type Master struct {
	// Your definitions here.
	MapTaskList    chan *Task
	ReduceTaskList chan *Task
	MapNum         int
	ReduceNum      int
	UniqueJobId    int
	MapTaskPool    map[int]*TCB
	ReduceTaskPool map[int]*TCB
	Phase          Phase
}

// Your code here -- RPC handlers for the worker to call.

// an example RPC handler.
//
// the RPC argument and reply types are defined in rpc.go.
func (m *Master) Example(args *ExampleArgs, reply *ExampleReply) error {
	reply.Y = args.X + 1
	return nil
}

func (m *Master) generateUniqueId() int {
	res := m.UniqueJobId
	m.UniqueJobId = res + 1
	return res
}

func (m *Master) generateMapTaskList(inputfiles []string) {
	for _, file := range inputfiles {
		taskId := m.generateUniqueId()
		mapTask := MakeTask(taskId, MapType, []string{file}, m.ReduceNum)
		m.MapTaskList <- mapTask
		m.MapTaskPool[mapTask.Id] = MakeTCB(mapTask) //初始化TaskState状态为Waiting
	}
	fmt.Printf("Generate map over!\n")
}

func (m *Master) generateReduceTaskList() {
	for i := 0; i < m.ReduceNum; i++ {
		taskId := m.generateUniqueId()
		reduceTask := MakeTask(taskId, ReduceType, TmpFileAssignHelper(i, "./"), m.ReduceNum)
		m.ReduceTaskList <- reduceTask
		m.ReduceTaskPool[reduceTask.Id] = MakeTCB(reduceTask) //初始化TaskState状态为Waiting
	}
	fmt.Printf("Generate reduce over!\n")
}

// RPC调用，worker调用他来获取任务 一定不要忘了格式 比如要返回error
func (m *Master) AllotTask(args *ExampleArgs, reply *Task) error {
	mutex.Lock()
	defer mutex.Unlock()
	switch m.Phase {
	case mapPhase:
		{
			if len(m.MapTaskList) == 0 {
				*reply = *MakeTask(m.generateUniqueId(), WaitingType, []string{}, m.ReduceNum)
			} else {
				*reply = *<-m.MapTaskList
				m.StartTask(reply.Id)
			}
		}
	case reducePhase:
		{
			if len(m.ReduceTaskList) == 0 {
				*reply = *MakeTask(m.generateUniqueId(), WaitingType, []string{}, m.ReduceNum)
			} else {
				*reply = *<-m.ReduceTaskList
				m.StartTask(reply.Id)
				m.getTaskTCB(reply.Id)
			}
		}
	case workDonePhase:
		{
			*reply = *MakeTask(m.generateUniqueId(), OverType, []string{}, m.ReduceNum)
		}
	}
	return nil
}

func (m *Master) masterStateHandle() {
	for m.Phase != workDonePhase {
		time.Sleep(time.Second)
		mutex.Lock()
		switch m.Phase {
		case mapPhase:
			{
				m.handleTimeOutMapTask()
				if m.isMapOver() {
					m.Phase = reducePhase
					m.generateReduceTaskList()
				}
			}
		case reducePhase:
			{
				m.handleTimeOutReduceTask()
				if m.isReduceOver() {
					m.Phase = workDonePhase
				}
			}
		}
		mutex.Unlock()
	}
}

func (m *Master) handleTimeOutMapTask() {
	for _, task := range m.MapTaskPool {
		if task.TaskState == TaskWorking && (time.Since(task.StartTime) > 2*time.Second) {
			task.TaskState = TaskWaiting
			m.MapTaskList <- task.Task
		}
	}
}

func (m *Master) handleTimeOutReduceTask() {
	for _, task := range m.ReduceTaskPool {
		if task.TaskState == TaskWorking && (time.Since(task.StartTime) > 2*time.Second) {
			task.TaskState = TaskWaiting
			m.ReduceTaskList <- task.Task
		}
	}
}

func (m *Master) isMapOver() bool {
	for _, task := range m.MapTaskPool {
		if task.TaskState == TaskWorking {
			return false
		}
	}
	return len(m.MapTaskList) == 0
}

func (m *Master) isReduceOver() bool {
	for _, task := range m.ReduceTaskPool {
		if task.TaskState == TaskWorking {
			return false
		}
	}
	return len(m.ReduceTaskList) == 0
}

func (m *Master) TaskDone(args *int, reply *ExampleReply) error {
	tcb, _ := m.getTaskTCB(*args)
	if tcb.TaskState == TaskDone {
		return nil
	}
	tcb.TaskState = TaskDone
	re := &ExampleReply{}
	*reply = *re
	return nil
}

func (m *Master) getTaskTCB(taskId int) (*TCB, error) {
	for _, mtcb := range m.MapTaskPool {
		if mtcb.Id == taskId {
			return mtcb, nil
		}
	}
	for _, mtcb := range m.ReduceTaskPool {
		if mtcb.Id == taskId {
			return mtcb, nil
		}
	}
	return nil, errors.New("No such Task,TaskID:" + strconv.Itoa(taskId))
}

func (m *Master) StartTask(args int) {
	tcb, err := m.getTaskTCB(args)
	if err != nil {
		fmt.Printf("%v\n", err)
	}
	err = tcb.StartTask()
	if err != nil {
		fmt.Printf("%v,TaskID:%v\n", err, args)
	}
}

// start a thread that listens for RPCs from worker.go
func (m *Master) Server() {
	rpc.Register(m)
	rpc.HandleHTTP()
	// l, e := net.Listen("tcp", ":1234")
	sockname := masterSock()
	os.Remove(sockname)
	l, e := net.Listen("unix", sockname)
	if e != nil {
		log.Fatal("listen error:", e)
	}
	go http.Serve(l, nil)
}

// main/mrmaster.go calls Done() periodically to find out
// if the entire job has finished.
func (m *Master) Done() bool {
	// Your code here.
	a := m.Phase == workDonePhase
	return a
}

// create a Master.
// main/mrmaster.go calls this function.
// nReduce is the number of reduce tasks to use.
func MakeMaster(files []string, nReduce int) *Master {
	m := Master{
		MapTaskList:    make(chan *Task, len(files)),
		ReduceTaskList: make(chan *Task, nReduce),
		MapNum:         len(files),
		ReduceNum:      nReduce,
		MapTaskPool:    make(map[int]*TCB),
		ReduceTaskPool: make(map[int]*TCB),
		Phase:          mapPhase,
	}
	// Your code here.
	m.generateMapTaskList(files)
	m.Server()
	go m.masterStateHandle()
	time.Sleep(100 * time.Second)
	return &m
}

func TmpFileAssignHelper(whichReduce int, tmpFileDirectoryName string) []string {
	var res []string
	rd, _ := ioutil.ReadDir(tmpFileDirectoryName)
	for _, fi := range rd {
		if strings.HasPrefix(fi.Name(), "mr-tmp") && strings.HasSuffix(fi.Name(), strconv.Itoa(whichReduce)) {
			res = append(res, fi.Name())
		}
	}
	return res
}
